package com.dm.core;

import com.dm.core.parser.SqlParser;
import com.dm.database.SQLType;
import com.dm.util.JedisUtil;
import com.dm.util.StringUtils;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author zy
 * 处理任务的管道
 */
public class Pipeline {

	private static final Logger LOGGER = LoggerFactory.getLogger(Pipeline.class);

	private BlockingQueue<RedisTask> bq = new LinkedBlockingQueue(100);

	/**
	 * sql解析器集合
	 */
	private List<SqlParser> sqlParsers;

	public Pipeline() {

	}

	public Pipeline(List<SqlParser> sqlParsers) {
		this.sqlParsers = sqlParsers;
		start();
	}

	public void process(SqlParserTask task) {
		sqlParsers.stream().forEach(s -> s.parser(task, this));
	}

	public void start() {
		PipelineContext.taskThreadPoolExecutor.execute(() -> {
			try {
				runWorker();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		});
	}

	private void runWorker() throws InterruptedException {
		RedisTask task;
		while ((task = getTask()) != null) {
			String key = task.getKey();
			SQLType sqlType = task.getSqlType();
			try {
				switch (sqlType) {
					case DELETE:
						if (JedisUtil.del(key) > 0) {
							LOGGER.info("Redis success to delete key:[{}]", key);
						} else {
							LOGGER.error("Redis fail to delete the key: [{}]", key);
						}
						break;
					case INSERT:
						if ("OK".equals(JedisUtil.setStringValue(key, task.getJson()))) {
							LOGGER.info("Redis success to insert task:[{}]", task);
						} else {
							LOGGER.error("Redis fail to insert the task: [{}]", task);
						}
						break;
					case UPDATE:
						String jsonString = JedisUtil.getStringValue(key);
						if (StringUtils.isBlank(jsonString)) {
							LOGGER.error("Redis fail to update task:[{}], redisTask = null", task);
						}
						JsonObject sourceJo = new JsonParser().parse(jsonString).getAsJsonObject();
						JsonObject targetJo = new JsonParser().parse(task.getJson()).getAsJsonObject();
						merge(targetJo, sourceJo);
						if ("OK".equals(JedisUtil.setStringValue(key, targetJo.toString()))) {
							LOGGER.info("Redis success to update task:[{}]", task);
						} else {
							LOGGER.info("Redis fail to update the task: [{}]", task);
						}
						break;
					default:
				}
			} catch (Exception e) {
				LOGGER.error(e.getMessage());
			}
		}
	}

	private void merge(JsonObject firstObj, JsonObject sourcerObj) {
		Set<Map.Entry<String, JsonElement>> entries = sourcerObj.entrySet();
		for (Map.Entry<String, JsonElement> e : entries) {
			String key = e.getKey();
			if (!firstObj.has(key)) {
				firstObj.add(key, e.getValue());
			}
		}
	}

	private RedisTask getTask() throws InterruptedException {
		while (true) {
			RedisTask take = bq.take();
			if (take != null) {
				return take;
			}
		}
	}

	public BlockingQueue getBq() {
		return bq;
	}
}